package com.hs.kfk.serializer;

import org.apache.kafka.clients.consumer.*;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * @Description: 基础班的消息消费方
 * @Author 胡尚
 * @Date: 2024/8/7 19:07
 */
public class UserConsumer {

    private final static String BOOTSTRAP_SERVER = "worker1:9092,worker2:9092,worker3:9092";
    private final static String TOPIC_NAME = "disTopic";
    private final static String CONSUMER_GROUP = "test";

    public static void main(String[] args) {
        // 设置消费者相关的属性
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVER);
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP);
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        // 配置value的反序列化类
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "com.hs.kfk.serializer.UserDeserializer");
        // 创建一个消息消费者对象
        Consumer<String, User> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Arrays.asList(TOPIC_NAME));
        while (true) {
            // 消费者拉取消息
            ConsumerRecords<String, User> consumerRecords = consumer.poll(Duration.ofMillis(100));

            // 处理消息
            for (ConsumerRecord<String, User> record : consumerRecords) {
                User message = record.value();
                System.out.println("message: " + message);
            }
            consumer.commitSync();
        }
    }
}
